package dm.ass1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TestFileInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import dm.ass1.io.ProfileWritable;
import dm.ass1.io.RatingWritable;

public class ProfilingDriver extends Configured implements Tool {
	
	public static final String PREFIX_USER = "U";
	public static final String PREFIX_ITEM = "I";
	public static final String PREFIX_PROPERTY = "prefix.value";
	public static final String TEMP_DIR = "/tmp/bigdata.temp";
	
	private boolean runProfilingJob(Configuration conf, String inputPath, String outputPath) throws Exception {
		FileSystem fs = FileSystem.getLocal(conf);
	    Path tempOutputPath = new Path(outputPath);
	    if (fs.exists(tempOutputPath)) {
	    	fs.delete(tempOutputPath,true);			
		}
	    		
		Job job = new Job(conf);
		job.setJarByClass(ProfilingDriver.class);
		FileInputFormat.addInputPath(job, new Path(inputPath));
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		SequenceFileOutputFormat.setOutputPath(job, tempOutputPath);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(ProfileWritable.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(RatingWritable.class);
		job.setMapperClass(ProfileMapper.class);
		job.setReducerClass(ProfileReducer.class);
		
		return job.waitForCompletion(true);		
	}
	
	private boolean runSplitOutputJob(Configuration conf, String inputPath, String output, String prefix) throws Exception {
		FileSystem fs = FileSystem.getLocal(conf);
	    Path outoutUsersPath = new Path(output);
	    if (fs.exists(outoutUsersPath)) {
	    	fs.delete(outoutUsersPath,true);			
		}
	    conf.set(ProfilingDriver.PREFIX_PROPERTY, prefix);
		Job job = new Job(conf);
		job.setJarByClass(ProfilingDriver.class);
		
		// input		
		SequenceFileInputFormat.addInputPath(job, new Path(inputPath));
		job.setInputFormatClass(SequenceFileInputFormat.class);
		
		// output
		job.setOutputFormatClass(TextOutputFormat.class);
		FileOutputFormat.setOutputPath(job, outoutUsersPath);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(ProfileWritable.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(ProfileWritable.class);
		job.setMapperClass(SplitOutputMapper.class);
		job.setReducerClass(SplitOutputReducer.class);
		
		return job.waitForCompletion(true);		
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = new Configuration();
		boolean profilingFinished = true; 
			
		profilingFinished = runProfilingJob(conf, args[0], ProfilingDriver.TEMP_DIR);
		
		if (profilingFinished) {
			runSplitOutputJob(conf, ProfilingDriver.TEMP_DIR, args[1], ProfilingDriver.PREFIX_USER);
			runSplitOutputJob(conf, ProfilingDriver.TEMP_DIR, args[2], ProfilingDriver.PREFIX_ITEM);
		}
		
		return profilingFinished?0:-1;

	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new ProfilingDriver(), args);
		System.exit(ret);
	}
}
